﻿using Grpc.Core;
using MagicOnion.Client.Internal;
using MagicOnion.Internal;
using MagicOnion.Internal.Buffers;
using MagicOnion.Serialization;
using System.Buffers;
using System.Threading.Channels;

namespace MagicOnion.Client;

public class StreamingHubClientOptions
{
    public string? Host { get; }
    public CallOptions CallOptions { get; }
    public IMagicOnionSerializerProvider SerializerProvider { get; }
    public IMagicOnionClientLogger Logger { get; }

    public TimeSpan? ClientHeartbeatInterval { get; }
    public TimeSpan? ClientHeartbeatTimeout { get; }
    public Action<ServerHeartbeatEvent>? OnServerHeartbeatReceived { get; }
    public Action<ClientHeartbeatEvent>? OnClientHeartbeatResponseReceived { get; }
    public TimeProvider? TimeProvider { get; }

    public StreamingHubClientOptions(string? host, CallOptions callOptions, IMagicOnionSerializerProvider serializerProvider, IMagicOnionClientLogger logger)
        : this(host, callOptions, serializerProvider, logger, default, default, default, default, default)
    {
    }

    public StreamingHubClientOptions(string? host, CallOptions callOptions, IMagicOnionSerializerProvider serializerProvider, IMagicOnionClientLogger logger, TimeSpan? clientHeartbeatInterval, TimeSpan? clientHeartbeatTimeout, Action<ServerHeartbeatEvent>? onServerHeartbeatReceived, Action<ClientHeartbeatEvent>? onClientHeartbeatResponseReceived,TimeProvider? timeProvider)
    {
        Host = host;
        CallOptions = callOptions;
        SerializerProvider = serializerProvider ?? throw new ArgumentNullException(nameof(serializerProvider));
        Logger = logger ?? throw new ArgumentNullException(nameof(logger));
        ClientHeartbeatInterval = clientHeartbeatInterval;
        ClientHeartbeatTimeout = clientHeartbeatTimeout;
        OnServerHeartbeatReceived = onServerHeartbeatReceived;
        OnClientHeartbeatResponseReceived = onClientHeartbeatResponseReceived;
        TimeProvider = timeProvider;
    }

    public static StreamingHubClientOptions CreateWithDefault(string? host = default, CallOptions callOptions = default, IMagicOnionSerializerProvider? serializerProvider = default, IMagicOnionClientLogger? logger = default)
        => new(host, callOptions, serializerProvider ?? MagicOnionSerializerProvider.Default, logger ?? NullMagicOnionClientLogger.Instance);

    public StreamingHubClientOptions WithHost(string? host)
        => new(host, CallOptions, SerializerProvider, Logger
            , ClientHeartbeatInterval, ClientHeartbeatTimeout, OnServerHeartbeatReceived, OnClientHeartbeatResponseReceived
            , TimeProvider
        );
    public StreamingHubClientOptions WithCallOptions(CallOptions callOptions)
        => new(Host, callOptions, SerializerProvider, Logger
            , ClientHeartbeatInterval, ClientHeartbeatTimeout, OnServerHeartbeatReceived, OnClientHeartbeatResponseReceived
            , TimeProvider
        );
    public StreamingHubClientOptions WithSerializerProvider(IMagicOnionSerializerProvider serializerProvider)
        => new(
            Host, CallOptions, serializerProvider, Logger
            , ClientHeartbeatInterval, ClientHeartbeatTimeout, OnServerHeartbeatReceived, OnClientHeartbeatResponseReceived
            , TimeProvider
        );
    public StreamingHubClientOptions WithLogger(IMagicOnionClientLogger logger)
        => new(Host, CallOptions, SerializerProvider, logger
            , ClientHeartbeatInterval, ClientHeartbeatTimeout, OnServerHeartbeatReceived, OnClientHeartbeatResponseReceived
            , TimeProvider
        );

    /// <summary>
    /// Sets a heartbeat interval. If a value is <see keyword="null"/>, the heartbeat from the client is disabled.
    /// </summary>
    /// <param name="interval"></param>
    /// <returns></returns>
    public StreamingHubClientOptions WithClientHeartbeatInterval(TimeSpan? interval)
        => new(Host, CallOptions, SerializerProvider, Logger
            , interval, ClientHeartbeatTimeout, OnServerHeartbeatReceived, OnClientHeartbeatResponseReceived
            , TimeProvider
        );

    /// <summary>
    /// Sets a heartbeat timeout period. If a value is <see keyword="null"/>, the client does not time out.
    /// </summary>
    /// <param name="timeout"></param>
    /// <returns></returns>
    public StreamingHubClientOptions WithClientHeartbeatTimeout(TimeSpan? timeout)
        => new(Host, CallOptions, SerializerProvider, Logger
            , ClientHeartbeatInterval, timeout, OnServerHeartbeatReceived, OnClientHeartbeatResponseReceived
            , TimeProvider
        );

    /// <summary>
    /// Sets a heartbeat callback. If additional metadata is provided by the server in the heartbeat message, this metadata is provided as an argument.
    /// </summary>
    /// <param name="onServerHeartbeatReceived"></param>
    /// <returns></returns>
    public StreamingHubClientOptions WithServerHeartbeatReceived(Action<ServerHeartbeatEvent>? onServerHeartbeatReceived)
        => new(Host, CallOptions, SerializerProvider, Logger
            , ClientHeartbeatInterval, ClientHeartbeatTimeout, onServerHeartbeatReceived, OnClientHeartbeatResponseReceived
            , TimeProvider
        );

    /// <summary>
    /// Sets a client heartbeat response callback.
    /// </summary>
    /// <param name="onClientHeartbeatResponseReceived"></param>
    /// <returns></returns>
    public StreamingHubClientOptions WithClientHeartbeatResponseReceived(Action<ClientHeartbeatEvent>? onClientHeartbeatResponseReceived)
        => new(Host, CallOptions, SerializerProvider, Logger
            , ClientHeartbeatInterval, ClientHeartbeatTimeout, OnServerHeartbeatReceived, onClientHeartbeatResponseReceived
            , TimeProvider
        );

    /// <summary>
    /// Sets a <see cref="TimeProvider"/>
    /// </summary>
    /// <param name="timeProvider"></param>
    /// <returns></returns>
    public StreamingHubClientOptions WithTimeProvider(TimeProvider timeProvider)
        => new(Host, CallOptions, SerializerProvider, Logger
            , ClientHeartbeatInterval, ClientHeartbeatTimeout, OnServerHeartbeatReceived, OnClientHeartbeatResponseReceived
            , timeProvider
        );
}

public abstract class StreamingHubClientBase<TStreamingHub, TReceiver> : IStreamingHubClient
    where TStreamingHub : IStreamingHub<TStreamingHub, TReceiver>
{
#pragma warning disable IDE1006 // Naming Styles
    const string StreamingHubVersionHeaderKey = "x-magiconion-streaminghub-version";
    const string StreamingHubVersionHeaderValue = "2";
#pragma warning restore IDE1006 // Naming Styles

    readonly CallInvoker callInvoker;
    readonly StreamingHubClientOptions options;
    readonly IMagicOnionClientLogger logger;
    readonly IMagicOnionSerializer messageSerializer;
    readonly Method<StreamingHubPayload, StreamingHubPayload> duplexStreamingConnectMethod;
    // {messageId, TaskCompletionSource}
    readonly Dictionary<int, IStreamingHubResponseTaskSource> responseFutures = new();
    readonly TaskCompletionSource<DisconnectionReason> waitForDisconnect = new();
    readonly CancellationTokenSource subscriptionCts = new();
    readonly Dictionary<int, SendOrPostCallback> postCallbackCache = new();

    int messageIdSequence = 0;
    bool disposed;
    bool disconnected;
    int cleanupSentinel = 0; // 0 = false, 1 = true

    readonly Channel<StreamingHubPayload> writerQueue = Channel.CreateUnbounded<StreamingHubPayload>(new UnboundedChannelOptions() { SingleReader = true, SingleWriter = false, AllowSynchronousContinuations = false });
    Task? writerTask;
    IClientStreamWriter<StreamingHubPayload> writer = default!;
    IAsyncStreamReader<StreamingHubPayload> reader = default!;

    StreamingHubClientHeartbeatManager heartbeatManager = default!;
    Task subscription = default!;

    protected readonly TReceiver receiver;

    protected StreamingHubClientBase(string serviceName, TReceiver receiver, CallInvoker callInvoker, StreamingHubClientOptions options)
    {
        this.callInvoker = callInvoker;
        this.receiver = receiver;
        this.options = options;
        this.logger = options.Logger;
        this.duplexStreamingConnectMethod = CreateConnectMethod(serviceName);
        this.messageSerializer = options.SerializerProvider.Create(MethodType.DuplexStreaming, null);
    }

    // call immediately after create.
    internal async Task __ConnectAndSubscribeAsync(CancellationToken connectAndSubscribeCancellationToken)
    {
        var syncContext = SynchronizationContext.Current; // capture SynchronizationContext.
        var callResult = callInvoker.AsyncDuplexStreamingCall(duplexStreamingConnectMethod, options.Host, options.CallOptions);

        this.writer = callResult.RequestStream;
        this.reader = callResult.ResponseStream;

        var cancelTcs = new TaskCompletionSource<bool>();
        using var cancelRegistration = connectAndSubscribeCancellationToken.Register(static cancelTcs =>
        {
            ((TaskCompletionSource<bool>)cancelTcs!).SetCanceled();
        }, cancelTcs);

        // Establish StreamingHub connection between the client and the server.
        Metadata.Entry? messageVersion;
        try
        {
            // The client can read the response headers before any StreamingHub's message.
            // MagicOnion.Server v4.0.x or before doesn't send any response headers. The client is incompatible with that versions.
            // NOTE: Grpc.Net:
            //           If the channel can not be connected, ResponseHeadersAsync will throw an exception.
            //       C-core:
            //           If the channel can not be connected, ResponseHeadersAsync will **return** an empty metadata.

            // ResponseHeadersAsync does not accept CancellationToken, so use TaskCompletionSource to wait for CancellationToken.
            var responseHeadersTask = callResult.ResponseHeadersAsync;
            var completedTask = await Task.WhenAny(responseHeadersTask, cancelTcs.Task).ConfigureAwait(false);

            // If cancellation is requested before the connection is established, the connection will be maintained, so callResult must be disposed.
            // After this, the only thing that passes connectAndSubscribeCancellationToken is MoveNext of Stream, so it doesn't need to be disposed.
            if (completedTask == cancelTcs.Task)
            {
                try
                {
                    connectAndSubscribeCancellationToken.ThrowIfCancellationRequested();
                }
                finally
                {
                    callResult.Dispose();
                }
                return;
            }

            var headers = responseHeadersTask.Result;
            messageVersion = headers.FirstOrDefault(x => x.Key == StreamingHubVersionHeaderKey);

            // Check message version of StreamingHub.
            if (messageVersion != null && messageVersion.Value != StreamingHubVersionHeaderValue)
            {
                throw new RpcException(new Status(StatusCode.Internal, $"The message version of StreamingHub mismatch between the client and the server. (ServerVersion={messageVersion?.Value}; Expected={StreamingHubVersionHeaderValue})"));
            }
        }
        catch (RpcException e)
        {
            throw new RpcException(e.Status, $"Failed to connect to StreamingHub '{duplexStreamingConnectMethod.ServiceName}'. ({e.Status})");
        }

        // Set up the Heartbeat Manager
        heartbeatManager = new StreamingHubClientHeartbeatManager(
            writerQueue.Writer,
            options.ClientHeartbeatInterval ?? TimeSpan.Zero /* Disable */,
            options.ClientHeartbeatTimeout ?? Timeout.InfiniteTimeSpan,
            options.OnServerHeartbeatReceived,
            options.OnClientHeartbeatResponseReceived,
            syncContext,
            options.TimeProvider ?? TimeProvider.System
        );

        // Activate the Heartbeat Manager if enabled in the options.
        var subscriptionToken = subscriptionCts.Token;
        if (options.ClientHeartbeatInterval is { } heartbeatInterval && heartbeatInterval > TimeSpan.Zero)
        {
            heartbeatManager.StartClientHeartbeatLoop();
            subscriptionToken = CancellationTokenSource.CreateLinkedTokenSource(heartbeatManager.TimeoutToken, subscriptionCts.Token).Token;
        }

        var firstMoveNextTask = reader.MoveNext(subscriptionToken);
        if (firstMoveNextTask.IsFaulted || messageVersion == null)
        {
            // NOTE: Grpc.Net:
            //           If an error is returned from `StreamingHub.Connect` method on a server-side,
            //           ResponseStream.MoveNext synchronously returns a task that is `IsFaulted = true`.
            //           `ConnectAsync` method should throw an exception here immediately.
            await firstMoveNextTask.ConfigureAwait(false);

            throw new RpcException(new Status(StatusCode.Internal, $"The request started successfully (StatusCode = OK), but the StreamingHub client has failed to negotiate with the server. ServerVersion is missing."));
        }

        this.subscription = StartSubscribe(syncContext, firstMoveNextTask, subscriptionToken);
    }

    // Helper methods to make building clients easy.
    protected void SetResultForResponse<TResponse>(object taskSource, ReadOnlyMemory<byte> data)
        => ((StreamingHubResponseTaskSource<TResponse>)taskSource).SetResult(Deserialize<TResponse>(data));
    protected void Serialize<T>(IBufferWriter<byte> writer, in T value)
        => messageSerializer.Serialize<T>(writer, value);
    protected T Deserialize<T>(ReadOnlyMemory<byte> data)
        => messageSerializer.Deserialize<T>(new ReadOnlySequence<byte>(data));

    protected abstract void OnClientResultEvent(int methodId, Guid messageId, ReadOnlyMemory<byte> data);
    protected abstract void OnResponseEvent(int methodId, object taskSource, ReadOnlyMemory<byte> data);
    protected abstract void OnBroadcastEvent(int methodId, ReadOnlyMemory<byte> data);

    static Method<StreamingHubPayload, StreamingHubPayload> CreateConnectMethod(string serviceName)
        => new (MethodType.DuplexStreaming, serviceName, "Connect", MagicOnionMarshallers.StreamingHubMarshaller, MagicOnionMarshallers.StreamingHubMarshaller);

    async Task StartSubscribe(SynchronizationContext? syncContext, Task<bool> firstMoveNext, CancellationToken subscriptionToken)
    {
        var disconnectionReason = new DisconnectionReason(DisconnectionType.CompletedNormally, null);
        writerTask = RunWriterLoopAsync(subscriptionToken);

        var reader = this.reader;
        try
        {
            var moveNext = firstMoveNext;
            while (await moveNext.ConfigureAwait(false)) // avoid Post to SyncContext(it losts one-frame per operation)
            {
                try
                {
                    ConsumeData(syncContext, reader.Current);
                }
                catch (Exception ex)
                {
                    const string msg = "An error occurred when consuming a received message, but the subscription is still alive.";
                    // log post on main thread.
                    if (syncContext != null)
                    {
                        syncContext.Post(s => logger.Error((Exception)s!, msg), ex);
                    }
                    else
                    {
                        logger.Error(ex, msg);
                    }
                }

                moveNext = reader.MoveNext(subscriptionToken);
            }
        }
        catch (Exception ex)
        {
            // When terminating by Heartbeat or DisposeAsync, a RpcException with a Status of Canceled is thrown.
            // If `ex.InnerException` is OperationCanceledException` and `subscriptionToken.IsCancellationRequested` is true, it is treated as a normal cancellation.
            if ((ex is OperationCanceledException oce) ||
                (ex is RpcException { InnerException: OperationCanceledException } && subscriptionToken.IsCancellationRequested))
            {
                if (heartbeatManager.TimeoutToken.IsCancellationRequested)
                {
                    disconnectionReason = new DisconnectionReason(DisconnectionType.TimedOut, ex);
                }
                return;
            }

            const string msg = "An error occurred while subscribing to messages.";
            // log post on main thread.
            if (syncContext != null)
            {
                syncContext.Post(s => logger.Error((Exception)s!, msg), ex);
            }
            else
            {
                logger.Error(ex, msg);
            }

            disconnectionReason = new DisconnectionReason(DisconnectionType.Faulted, ex);
        }
        finally
        {
            disconnected = true;

            try
            {
#if !UNITY_WEBGL
                // set syncContext before await
                // NOTE: If restore SynchronizationContext in WebGL environment, a continuation will not be executed inline and will be stuck.
                if (syncContext != null && SynchronizationContext.Current == null)
                {
                    SynchronizationContext.SetSynchronizationContext(syncContext);
                }
#endif
                await heartbeatManager.DisposeAsync().ConfigureAwait(false);
                await CleanupAsync(false).ConfigureAwait(false);
            }
            finally
            {
                waitForDisconnect.TrySetResult(disconnectionReason);
            }
        }
    }

    void ConsumeData(SynchronizationContext? syncContext, StreamingHubPayload payload)
    {
        var messageReader = new StreamingHubClientMessageReader(payload.Memory);
        switch (messageReader.ReadMessageType())
        {
            case StreamingHubMessageType.Broadcast:
                ProcessBroadcast(syncContext, payload, ref messageReader);
                break;
            case StreamingHubMessageType.Response:
                ProcessResponse(syncContext, payload, ref messageReader);
                break;
            case StreamingHubMessageType.ResponseWithError:
                ProcessResponseWithError(syncContext, payload, ref messageReader);
                break;
            case StreamingHubMessageType.ClientResultRequest:
                ProcessClientResultRequest(syncContext, payload, ref messageReader);
                break;
            case StreamingHubMessageType.ServerHeartbeat:
                heartbeatManager.ProcessServerHeartbeat(payload);
                break;
            case StreamingHubMessageType.ClientHeartbeatResponse:
                heartbeatManager.ProcessClientHeartbeatResponse(payload);
                break;
        }
    }

    void ProcessBroadcast(SynchronizationContext? syncContext, StreamingHubPayload payload, ref StreamingHubClientMessageReader messageReader)
    {
        if (syncContext is null)
        {
            var message = messageReader.ReadBroadcastMessage();
            OnBroadcastEvent(message.MethodId, message.Body);
            StreamingHubPayloadPool.Shared.Return(payload);
        }
        else
        {
            var (methodId, consumed) = messageReader.ReadBroadcastMessageMethodId();
            if (!postCallbackCache.TryGetValue(methodId, out var postCallback))
            {
                // Create and cache a callback delegate capturing `this` and the header size.
                postCallback = postCallbackCache[methodId] = CreateBroadcastCallback(methodId, consumed);
            }
            syncContext.Post(postCallback, payload);
        }
    }

    SendOrPostCallback CreateBroadcastCallback(int methodId, int consumed)
    {
        return (s) =>
        {
            var p = (StreamingHubPayload)s!;
            this.OnBroadcastEvent(methodId, p.Memory.Slice(consumed));
            StreamingHubPayloadPool.Shared.Return(p);
        };
    }

    void ProcessResponse(SynchronizationContext? syncContext, StreamingHubPayload payload, ref StreamingHubClientMessageReader messageReader)
    {
        var message = messageReader.ReadResponseMessage();

        IStreamingHubResponseTaskSource? future;
        lock (responseFutures)
        {
            if (!responseFutures.Remove(message.MessageId, out future))
            {
                return;
            }
        }

        try
        {
            OnResponseEvent(message.MethodId, future, message.Body);
            StreamingHubPayloadPool.Shared.Return(payload);
        }
        catch (Exception ex)
        {
            if (!future.TrySetException(ex))
            {
                throw;
            }
        }
    }

    void ProcessResponseWithError(SynchronizationContext? syncContext, StreamingHubPayload payload, ref StreamingHubClientMessageReader messageReader)
    {
        var message = messageReader.ReadResponseWithErrorMessage();

        IStreamingHubResponseTaskSource? future;
        lock (responseFutures)
        {
            if (!responseFutures.Remove(message.MessageId, out future))
            {
                return;
            }
        }

        RpcException ex;
        if (string.IsNullOrWhiteSpace(message.Error))
        {
            ex = new RpcException(new Status((StatusCode)message.StatusCode, message.Detail ?? string.Empty));
        }
        else
        {
            ex = new RpcException(new Status((StatusCode)message.StatusCode, message.Detail ?? string.Empty), message.Detail + Environment.NewLine + message.Error);
        }

        future.TrySetException(ex);
        StreamingHubPayloadPool.Shared.Return(payload);
    }

    void ProcessClientResultRequest(SynchronizationContext? syncContext, StreamingHubPayload payload, ref StreamingHubClientMessageReader messageReader)
    {
        var message = messageReader.ReadClientResultRequestMessage();
        if (syncContext is null)
        {
            OnClientResultEvent(message.MethodId, message.ClientResultRequestMessageId, message.Body);
            StreamingHubPayloadPool.Shared.Return(payload);
        }
        else
        {
            var tuple = Tuple.Create(this, message.MethodId, message.ClientResultRequestMessageId, message.Body, payload);
            syncContext.Post(static state =>
            {
                var t = (Tuple<StreamingHubClientBase<TStreamingHub, TReceiver>, int, Guid, ReadOnlyMemory<byte>, StreamingHubPayload>)state!;
                t.Item1.OnClientResultEvent(t.Item2, t.Item3, t.Item4);
                StreamingHubPayloadPool.Shared.Return(t.Item5);
            }, tuple);
        }
    }

    async Task RunWriterLoopAsync(CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                if (await writerQueue.Reader.WaitToReadAsync(default).ConfigureAwait(false))
                {
                    while (writerQueue.Reader.TryRead(out var payload))
                    {
                        cancellationToken.ThrowIfCancellationRequested();
                        await writer.WriteAsync(payload).ConfigureAwait(false);
                    }
                }
            }
        }
        catch { /* Ignore */ }
    }

    protected Task<TResponse> WriteMessageFireAndForgetTaskAsync<TRequest, TResponse>(int methodId, TRequest message)
        => WriteMessageFireAndForgetValueTaskOfTAsync<TRequest, TResponse>(methodId, message).AsTask();

    protected ValueTask<TResponse> WriteMessageFireAndForgetValueTaskOfTAsync<TRequest, TResponse>(int methodId, TRequest message)
    {
        ThrowIfDisposed();
        ThrowIfDisconnected();

        var v = BuildRequestMessage(methodId, message);
        _ = writerQueue.Writer.TryWrite(v);

        return default;
    }

    protected ValueTask WriteMessageFireAndForgetValueTaskAsync<TRequest, TResponse>(int methodId, TRequest message)
    {
        WriteMessageFireAndForgetValueTaskOfTAsync<TRequest, TResponse>(methodId, message);
        return default;
    }

    protected Task<TResponse> WriteMessageWithResponseTaskAsync<TRequest, TResponse>(int methodId, TRequest message)
        => WriteMessageWithResponseValueTaskOfTAsync<TRequest, TResponse>(methodId, message).AsTask();

    protected ValueTask<TResponse> WriteMessageWithResponseValueTaskOfTAsync<TRequest, TResponse>(int methodId, TRequest message)
    {
        ThrowIfDisposed();
        ThrowIfDisconnected();

        var mid = Interlocked.Increment(ref messageIdSequence);

        var taskSource = StreamingHubResponseTaskSourcePool<TResponse>.Shared.RentOrCreate();
        lock (responseFutures)
        {
            responseFutures[mid] = taskSource;
        }

        var v = BuildRequestMessage(methodId, mid, message);
        if (!writerQueue.Writer.TryWrite(v))
        {
            // If the channel writer is closed, it is likely that the connection has already been disconnected.
            ThrowIfDisconnected();
        }

        return new ValueTask<TResponse>(taskSource, taskSource.Version); // wait until server return response(or error). if connection was closed, throws cancellation from DisposeAsyncCore.
    }

    protected ValueTask WriteMessageWithResponseValueTaskAsync<TRequest, TResponse>(int methodId, TRequest message)
    {
        ThrowIfDisposed();
        ThrowIfDisconnected();

        var mid = Interlocked.Increment(ref messageIdSequence);

        var taskSource = StreamingHubResponseTaskSourcePool<TResponse>.Shared.RentOrCreate();
        lock (responseFutures)
        {
            responseFutures[mid] = taskSource;
        }

        var v = BuildRequestMessage(methodId, mid, message);
        if (!writerQueue.Writer.TryWrite(v))
        {
            // If the channel writer is closed, it is likely that the connection has already been disconnected.
            ThrowIfDisconnected();
        }

        return new ValueTask(taskSource, taskSource.Version); // wait until server return response(or error). if connection was closed, throws cancellation from DisposeAsyncCore.
    }

    protected void AwaitAndWriteClientResultResponseMessage(int methodId, Guid clientResultMessageId, Task task)
        => AwaitAndWriteClientResultResponseMessage(methodId, clientResultMessageId, new ValueTask(task));

    protected async void AwaitAndWriteClientResultResponseMessage(int methodId, Guid clientResultMessageId, ValueTask task)
    {
        try
        {
            await task.ConfigureAwait(false);
            await WriteClientResultResponseMessageAsync(methodId, clientResultMessageId, MessagePack.Nil.Default).ConfigureAwait(false);
        }
        catch (Exception e)
        {
            await WriteClientResultResponseMessageForErrorAsync(methodId, clientResultMessageId, e).ConfigureAwait(false);
        }
    }

    protected void AwaitAndWriteClientResultResponseMessage<T>(int methodId, Guid clientResultMessageId, Task<T> task)
        => AwaitAndWriteClientResultResponseMessage(methodId, clientResultMessageId, new ValueTask<T>(task));

    protected async void AwaitAndWriteClientResultResponseMessage<T>(int methodId, Guid clientResultMessageId, ValueTask<T> task)
    {
        try
        {
            var result = await task.ConfigureAwait(false);
            await WriteClientResultResponseMessageAsync(methodId, clientResultMessageId, result).ConfigureAwait(false);
        }
        catch (Exception e)
        {
            await WriteClientResultResponseMessageForErrorAsync(methodId, clientResultMessageId, e).ConfigureAwait(false);
        }
    }

    protected async void WriteClientResultResponseMessageForError(int methodId, Guid clientResultMessageId, Exception ex)
    {
        try
        {
            await WriteClientResultResponseMessageForErrorAsync(methodId, clientResultMessageId, ex).ConfigureAwait(false);
        }
        catch
        {
            // Ignore Exception
        }
    }

    protected Task WriteClientResultResponseMessageAsync<T>(int methodId, Guid clientResultMessageId, T result)
    {
        var v = BuildClientResultResponseMessage(methodId, clientResultMessageId, result);
        _ = writerQueue.Writer.TryWrite(v);
        return Task.CompletedTask;
    }

    protected Task WriteClientResultResponseMessageForErrorAsync(int methodId, Guid clientResultMessageId, Exception ex)
    {
        var statusCode = ex is RpcException rpcException
            ? rpcException.StatusCode
            : StatusCode.Internal;

        var v = BuildClientResultResponseMessageForError(methodId, clientResultMessageId, (int)statusCode, ex.Message, ex);
        _ = writerQueue.Writer.TryWrite(v);

        return Task.CompletedTask;
    }

    void ThrowIfDisconnected()
    {
        if (disconnected)
        {
            throw new RpcException(new Status(StatusCode.Unavailable, $"The StreamingHubClient has already been disconnected from the server."));
        }
    }

    void ThrowIfDisposed()
    {
        if (disposed)
        {
            throw new ObjectDisposedException("StreamingHubClient", $"The StreamingHubClient has already been disconnected from the server.");
        }
    }

    public Task WaitForDisconnect()
        => ((IStreamingHubClient)this).WaitForDisconnectAsync();

    Task<DisconnectionReason> IStreamingHubClient.WaitForDisconnectAsync()
        => waitForDisconnect.Task;

    public async Task DisposeAsync()
    {
        if (disposed) return;
        disposed = true;
        await CleanupAsync(true).ConfigureAwait(false);
    }

    async ValueTask CleanupAsync(bool waitSubscription)
    {
        if (Interlocked.CompareExchange(ref cleanupSentinel, 1, 0) != 0)
        {
            return;
        }

        if (writer == null) return;
        try
        {
            writerQueue.Writer.Complete();
            await writer.CompleteAsync().ConfigureAwait(false);
        }
        catch { } // ignore error?
        finally
        {
            subscriptionCts.Cancel();
            try
            {
                if (waitSubscription)
                {
                    if (subscription != null)
                    {
                        await subscription.ConfigureAwait(false);
                    }
                }

                // cleanup completion
                List<Exception>? aggregateException = null;
                foreach (var item in responseFutures)
                {
                    try
                    {
                        (item.Value as IStreamingHubResponseTaskSource).TrySetCanceled();
                    }
                    catch (Exception ex)
                    {
                        if (!(ex is OperationCanceledException))
                        {
                            aggregateException ??= new List<Exception>();
                            aggregateException.Add(ex);
                        }
                    }
                }
                if (aggregateException != null)
                {
                    throw new AggregateException(aggregateException);
                }
            }
            catch (Exception ex)
            {
                if (!(ex is OperationCanceledException))
                {
                    throw;
                }
            }
        }
        subscriptionCts.Dispose();
    }

    StreamingHubPayload BuildRequestMessage<T>(int methodId, T message)
    {
        using var buffer = ArrayPoolBufferWriter.RentThreadStaticWriter();
        StreamingHubMessageWriter.WriteRequestMessageVoid(buffer, methodId, message, messageSerializer);
        return StreamingHubPayloadPool.Shared.RentOrCreate(buffer.WrittenSpan);
    }

    StreamingHubPayload BuildRequestMessage<T>(int methodId, int messageId, T message)
    {
        using var buffer = ArrayPoolBufferWriter.RentThreadStaticWriter();
        StreamingHubMessageWriter.WriteRequestMessage(buffer, methodId, messageId, message, messageSerializer);
        return StreamingHubPayloadPool.Shared.RentOrCreate(buffer.WrittenSpan);
    }

    StreamingHubPayload BuildClientResultResponseMessage<T>(int methodId, Guid messageId, T response)
    {
        using var buffer = ArrayPoolBufferWriter.RentThreadStaticWriter();
        StreamingHubMessageWriter.WriteClientResultResponseMessage(buffer, methodId, messageId, response, messageSerializer);
        return StreamingHubPayloadPool.Shared.RentOrCreate(buffer.WrittenSpan);
    }

    StreamingHubPayload BuildClientResultResponseMessageForError(int methodId, Guid messageId, int statusCode, string detail, Exception? ex)
    {
        using var buffer = ArrayPoolBufferWriter.RentThreadStaticWriter();
        StreamingHubMessageWriter.WriteClientResultResponseMessageForError(buffer, methodId, messageId, statusCode, detail, ex, messageSerializer);
        return StreamingHubPayloadPool.Shared.RentOrCreate(buffer.WrittenSpan);
    }
}
